/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.util.Map;

import jvs.peer.util.Log;

/**
 * @author qiangli
 *
 */
public class MessageDispatcher implements Runnable {
	protected Map<Object, Channel> channels = null;
	
	protected MessageDriver driver = null;

	private ChannelManager man = null;
	/**
	 * 
	 */
	public MessageDispatcher(ChannelManager man, Map<Object, Channel> channels, MessageDriver driver) {
		this.man = man;
		this.channels = channels;
		this.driver = driver;
	}
	
	protected void dispatch() throws Exception, InterruptedException {
		Message m = receive();
		int mode = m.getHeader().getMode();
		String peer = m.getHeader().getPeer();
		String tid = m.getHeader().getThread();
		String key = ((mode == MessageConstant.response)?"":peer)+"|"+tid;
		
		int cmd  = m.getIoscmd();
		Channel ch = null;
		switch(cmd) {
		case MessageConstant.open:
			ch = man.createServerChannel(tid, peer);
			process(key, ch, m);
			break;
		case MessageConstant.data:
		case MessageConstant.close:
			ch = channels.get(key);
			process(key, ch, m);
			break;
		default:
			throw new RuntimeException("not supported: "+cmd+" "+key);
		}
	}
	
	private void process(final String key, Channel ch, Message m) throws IOException {
		if (ch == null) {
			return;
		}
		try {
			ch.receive(m);
		} catch (Exception e) {
			channels.remove(key);
			
			ch.close();
		
			//
			e.printStackTrace();
		}
	}
	
	protected Message receive() throws Exception {
		Message m = driver.receive();
		if (Log.DEBUG) {
			Log.log(this, m.getHeader().getThread()+":"+m.getHeader().getPeer());
		}
		return m;
	}

	public void run() {
		while(true) {
			try {
				dispatch();
			} catch (Exception e) {
				//log.fine(e.getMessage());	
				e.printStackTrace();
				try {
					Thread.sleep(10000);
				} catch (InterruptedException e1) {
					//
				}
			}
		}
	}
}
